/*
 * Copyright (c) 2021 Contributors to the Eclipse Foundation
 * 
 * See the NOTICE file(s) distributed with this work for additional
 * information regarding copyright ownership.
 * 
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0
 * 
 * SPDX-License-Identifier: EPL-2.0
 */
package com.ibm.ims.svt.clients;

import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;

/**
 * This class is used as an asynchronous callback for an MQTT producer.
 * 
 *
 */
public class MqttProducerListener implements MqttCallback {
	
	private ArrayList<String> sentMsgs = new ArrayList<String>();
	private ConcurrentHashMap<Integer, String> idMap = null;
	private MqttProducerClient testClass = null;
	private boolean storeSentMsgs = false;
	
	/**
	 * 
	 * @param testClass - provides a callback to the client
	 * @param idMap - the list of ID's of the tokens from the published messages
	 * @param storeSentMsgs - if true, store the sent messages for the ID's that are completed
	 */
	public MqttProducerListener(MqttProducerClient testClass, ConcurrentHashMap<Integer, String> idMap, boolean storeSentMsgs)
	{
		this.testClass = testClass;
		this.idMap = idMap;
		this.storeSentMsgs = storeSentMsgs;
	}

	/**
	 * If connection is lost get the client to reconnect
	 */
	public void connectionLost(Throwable arg0) {
		
		
		if(testClass!= null && testClass.trace!= null && testClass.trace.traceOn())
		{
			testClass.trace.trace("MQTT connection was lost. Message=" + arg0.getMessage());
		}
		
		if(! testClass.isreconnecting())
		{
			if(testClass!= null && testClass.trace!= null && testClass.trace.traceOn())
			{
				testClass.trace.trace("Calling reconnect");
			}
			
			testClass.resetConnections();
		}
		else
		{
			if(testClass!= null && testClass.trace!= null && testClass.trace.traceOn())
			{
				testClass.trace.trace("Already reconnecting, just need to wait");
			}
		}
		
	}

	/**
	 * This is called asynchronously when delivery is completed
	 */
	public void deliveryComplete(IMqttDeliveryToken arg0) {

		String msgContents = null;
		if(arg0.isComplete())
		{
			if(storeSentMsgs)
			{ 
				// Get the message from the correlated token ID
				msgContents = idMap.get(arg0.getMessageId());
				if(msgContents != null)
				{
					// Store the message contents
					sentMsgs.add(msgContents);					
					idMap.remove(arg0.getMessageId());
				}
				else
				{
					//try again in case we were just too quick as we can sometimes get the completion before we 
					// have managed to store the token in the client
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						
					}
					msgContents = idMap.get(arg0.getMessageId());
					if(msgContents != null)
					{
						sentMsgs.add(msgContents);					
						idMap.remove(arg0.getMessageId());
					}
				}
			}
			if(testClass!= null && testClass.trace!= null && testClass.trace.traceOn())
			{
				testClass.trace.trace("MQTT delivery was complete for messageID " + arg0.getMessageId());
			}
				
			
		}
		else
		{
			if(testClass!= null && testClass.trace!= null && testClass.trace.traceOn())
			{
				testClass.trace.trace("MQTT delivery was not complete for messageID " + arg0.getMessageId());
			}
		}
		
	}

	/**
	 * This callback is only being used for producers so ignore
	 */
	public void messageArrived(String arg0, MqttMessage message)
			throws Exception {
	// do nothing here
		
	}

	/**
	 * Get the list of store messages (if configured to store them)
	 * @return
	 */
	public ArrayList<String> getSentMsgs()
	{
		return sentMsgs;
	}

}
